package com.reactim.message.performance;

import com.reactim.message.connection.ConnectionManager;
import com.reactim.message.connection.ConnectionInfo;
import com.reactim.message.metrics.ConnectionPerformanceMetrics;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 性能测试服务
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class PerformanceTestService {

    private final ConnectionManager connectionManager;
    private final ConnectionPerformanceMetrics performanceMetrics;

    /**
     * 执行完整的性能测试套件
     */
    public PerformanceTestResult runFullPerformanceTest() {
        log.info("开始执行完整性能测试套件...");
        
        PerformanceTestResult result = new PerformanceTestResult();
        
        try {
            // 1. 基础延迟测试
            log.info("执行基础延迟测试...");
            LatencyTestResult latencyResult = runLatencyTest();
            result.setLatencyResult(latencyResult);
            
            // 2. 吞吐量测试
            log.info("执行吞吐量测试...");
            ThroughputTestResult throughputResult = runThroughputTest();
            result.setThroughputResult(throughputResult);
            
            // 3. 并发测试
            log.info("执行并发测试...");
            ConcurrencyTestResult concurrencyResult = runConcurrencyTest();
            result.setConcurrencyResult(concurrencyResult);
            
            // 4. 负载测试
            log.info("执行负载测试...");
            LoadTestResult loadResult = runLoadTest();
            result.setLoadResult(loadResult);
            
            // 5. 压力测试
            log.info("执行压力测试...");
            StressTestResult stressResult = runStressTest();
            result.setStressResult(stressResult);
            
            log.info("完整性能测试套件执行完成");
            
        } catch (Exception e) {
            log.error("性能测试执行失败", e);
            throw new RuntimeException("性能测试失败", e);
        }
        
        return result;
    }

    /**
     * 延迟测试
     */
    public LatencyTestResult runLatencyTest() {
        log.info("开始延迟测试...");
        
        int testCount = 1000;
        List<Long> storeLatencies = new ArrayList<>();
        List<Long> retrieveLatencies = new ArrayList<>();
        List<Long> removeLatencies = new ArrayList<>();
        
        // 测试存储延迟
        for (int i = 0; i < testCount; i++) {
            long userId = (long) i;
            ConnectionInfo connectionInfo = createTestConnectionInfo(userId);
            
            long startTime = System.nanoTime();
            connectionManager.storeConnection(userId, null, connectionInfo).block();
            long endTime = System.nanoTime();
            
            storeLatencies.add((endTime - startTime) / 1_000_000); // 转换为毫秒
        }
        
        // 测试检索延迟
        for (int i = 0; i < testCount; i++) {
            long userId = (long) i;
            
            long startTime = System.nanoTime();
            connectionManager.getConnection(userId).block();
            long endTime = System.nanoTime();
            
            retrieveLatencies.add((endTime - startTime) / 1_000_000);
        }
        
        // 测试移除延迟
        for (int i = 0; i < testCount; i++) {
            long userId = (long) i;
            
            long startTime = System.nanoTime();
            connectionManager.removeConnection(userId).block();
            long endTime = System.nanoTime();
            
            removeLatencies.add((endTime - startTime) / 1_000_000);
        }
        
        return LatencyTestResult.builder()
                .testCount(testCount)
                .storeLatencies(calculateLatencyStats(storeLatencies))
                .retrieveLatencies(calculateLatencyStats(retrieveLatencies))
                .removeLatencies(calculateLatencyStats(removeLatencies))
                .build();
    }

    /**
     * 吞吐量测试
     */
    public ThroughputTestResult runThroughputTest() {
        log.info("开始吞吐量测试...");
        
        int[] testSizes = {1000, 5000, 10000};
        List<ThroughputMeasurement> measurements = new ArrayList<>();
        
        for (int testSize : testSizes) {
            log.info("测试规模: {} 操作", testSize);
            
            // 存储操作吞吐量
            long startTime = System.currentTimeMillis();
            
            Flux.range(0, testSize)
                .flatMap(i -> {
                    long userId = (long) i;
                    ConnectionInfo connectionInfo = createTestConnectionInfo(userId);
                    return connectionManager.storeConnection(userId, null, connectionInfo);
                }, 100) // 限制并发数
                .collectList()
                .block(Duration.ofMinutes(5));
            
            long endTime = System.currentTimeMillis();
            long duration = endTime - startTime;
            double throughput = (double) testSize / (duration / 1000.0);
            
            measurements.add(ThroughputMeasurement.builder()
                    .operationType("STORE")
                    .operationCount(testSize)
                    .durationMs(duration)
                    .throughput(throughput)
                    .build());
            
            // 清理数据
            cleanupTestData(testSize);
        }
        
        return ThroughputTestResult.builder()
                .measurements(measurements)
                .build();
    }

    /**
     * 并发测试
     */
    public ConcurrencyTestResult runConcurrencyTest() throws InterruptedException {
        log.info("开始并发测试...");
        
        int[] threadCounts = {5, 10, 20, 50};
        int operationsPerThread = 100;
        List<ConcurrencyMeasurement> measurements = new ArrayList<>();
        
        for (int threadCount : threadCounts) {
            log.info("测试并发数: {} 线程", threadCount);
            
            ExecutorService executor = Executors.newFixedThreadPool(threadCount);
            CountDownLatch latch = new CountDownLatch(threadCount);
            AtomicLong successCount = new AtomicLong(0);
            AtomicLong errorCount = new AtomicLong(0);
            
            long startTime = System.currentTimeMillis();
            
            for (int t = 0; t < threadCount; t++) {
                final int threadId = t;
                executor.submit(() -> {
                    try {
                        for (int i = 0; i < operationsPerThread; i++) {
                            long userId = threadId * operationsPerThread + i;
                            ConnectionInfo connectionInfo = createTestConnectionInfo(userId);
                            
                            try {
                                connectionManager.storeConnection(userId, null, connectionInfo)
                                    .block(Duration.ofSeconds(10));
                                successCount.incrementAndGet();
                            } catch (Exception e) {
                                errorCount.incrementAndGet();
                            }
                        }
                    } finally {
                        latch.countDown();
                    }
                });
            }
            
            latch.await(120, TimeUnit.SECONDS);
            executor.shutdown();
            
            long endTime = System.currentTimeMillis();
            long duration = endTime - startTime;
            int totalOperations = threadCount * operationsPerThread;
            double throughput = (double) totalOperations / (duration / 1000.0);
            
            measurements.add(ConcurrencyMeasurement.builder()
                    .threadCount(threadCount)
                    .operationsPerThread(operationsPerThread)
                    .totalOperations(totalOperations)
                    .successCount(successCount.get())
                    .errorCount(errorCount.get())
                    .durationMs(duration)
                    .throughput(throughput)
                    .build());
            
            // 清理数据
            cleanupTestData(totalOperations);
        }
        
        return ConcurrencyTestResult.builder()
                .measurements(measurements)
                .build();
    }

    /**
     * 负载测试
     */
    public LoadTestResult runLoadTest() {
        log.info("开始负载测试...");
        
        int maxConnections = 50000;
        int batchSize = 1000;
        List<LoadMeasurement> measurements = new ArrayList<>();
        
        for (int currentLoad = batchSize; currentLoad <= maxConnections; currentLoad += batchSize) {
            log.info("测试负载: {} 连接", currentLoad);
            
            long startTime = System.currentTimeMillis();
            
            // 创建连接
            Flux.range(0, currentLoad)
                .flatMap(i -> {
                    long userId = (long) i;
                    ConnectionInfo connectionInfo = createTestConnectionInfo(userId);
                    return connectionManager.storeConnection(userId, null, connectionInfo);
                }, 200) // 限制并发数
                .collectList()
                .block(Duration.ofMinutes(10));
            
            long createTime = System.currentTimeMillis() - startTime;
            
            // 测试查询性能
            startTime = System.currentTimeMillis();
            Long onlineCount = connectionManager.getOnlineUserCount().block();
            long queryTime = System.currentTimeMillis() - startTime;
            
            // 测试随机访问性能
            startTime = System.currentTimeMillis();
            for (int i = 0; i < 100; i++) {
                long randomUserId = (long) (Math.random() * currentLoad);
                connectionManager.getConnection(randomUserId).block();
            }
            long randomAccessTime = System.currentTimeMillis() - startTime;
            
            measurements.add(LoadMeasurement.builder()
                    .connectionCount(currentLoad)
                    .createTimeMs(createTime)
                    .queryTimeMs(queryTime)
                    .randomAccessTimeMs(randomAccessTime)
                    .actualOnlineCount(onlineCount != null ? onlineCount : 0)
                    .build());
            
            // 清理数据
            cleanupTestData(currentLoad);
            
            // 如果性能下降太多，停止测试
            if (createTime > 60000) { // 超过1分钟
                log.warn("创建时间过长，停止负载测试");
                break;
            }
        }
        
        return LoadTestResult.builder()
                .measurements(measurements)
                .build();
    }

    /**
     * 压力测试
     */
    public StressTestResult runStressTest() {
        log.info("开始压力测试...");
        
        int maxConnections = 100000;
        int incrementSize = 10000;
        List<StressMeasurement> measurements = new ArrayList<>();
        
        for (int load = incrementSize; load <= maxConnections; load += incrementSize) {
            log.info("压力测试负载: {} 连接", load);
            
            try {
                long startTime = System.currentTimeMillis();
                
                // 尝试创建大量连接
                Flux.range(0, load)
                    .flatMap(i -> {
                        long userId = (long) i;
                        ConnectionInfo connectionInfo = createTestConnectionInfo(userId);
                        return connectionManager.storeConnection(userId, null, connectionInfo)
                            .onErrorReturn(false); // 忽略错误，继续执行
                    }, 500) // 高并发
                    .collectList()
                    .block(Duration.ofMinutes(15));
                
                long duration = System.currentTimeMillis() - startTime;
                
                // 检查系统状态
                Long actualCount = connectionManager.getOnlineUserCount().block();
                
                measurements.add(StressMeasurement.builder()
                        .targetLoad(load)
                        .actualLoad(actualCount != null ? actualCount : 0)
                        .durationMs(duration)
                        .success(true)
                        .build());
                
                // 清理数据
                cleanupTestData(load);
                
            } catch (Exception e) {
                log.error("压力测试在负载 {} 时失败", load, e);
                
                measurements.add(StressMeasurement.builder()
                        .targetLoad(load)
                        .actualLoad(0)
                        .durationMs(0)
                        .success(false)
                        .errorMessage(e.getMessage())
                        .build());
                
                break; // 系统已经无法承受，停止测试
            }
        }
        
        return StressTestResult.builder()
                .measurements(measurements)
                .build();
    }

    /**
     * 计算延迟统计信息
     */
    private LatencyStats calculateLatencyStats(List<Long> latencies) {
        if (latencies.isEmpty()) {
            return LatencyStats.builder().build();
        }
        
        latencies.sort(Long::compareTo);
        
        double avg = latencies.stream().mapToLong(Long::longValue).average().orElse(0.0);
        long min = latencies.get(0);
        long max = latencies.get(latencies.size() - 1);
        long p50 = latencies.get((int) (latencies.size() * 0.5));
        long p95 = latencies.get((int) (latencies.size() * 0.95));
        long p99 = latencies.get((int) (latencies.size() * 0.99));
        
        return LatencyStats.builder()
                .average(avg)
                .min(min)
                .max(max)
                .p50(p50)
                .p95(p95)
                .p99(p99)
                .build();
    }

    /**
     * 创建测试连接信息
     */
    private ConnectionInfo createTestConnectionInfo(Long userId) {
        ConnectionInfo connectionInfo = new ConnectionInfo();
        connectionInfo.setUserId(userId);
        connectionInfo.setDeviceInfo("TestDevice-" + userId);
        connectionInfo.setClientVersion("1.0.0");
        connectionInfo.setTimestamp(System.currentTimeMillis());
        connectionInfo.setInstanceId("test-instance");
        return connectionInfo;
    }

    /**
     * 清理测试数据
     */
    private void cleanupTestData(int count) {
        try {
            Flux.range(0, count)
                .flatMap(i -> connectionManager.removeConnection((long) i)
                    .onErrorReturn(false))
                .collectList()
                .block(Duration.ofMinutes(5));
        } catch (Exception e) {
            log.warn("清理测试数据失败", e);
        }
    }

    // 数据结构定义
    @Data
    @Builder
    @NoArgsConstructor
    @AllArgsConstructor
    public static class PerformanceTestResult {
        private LatencyTestResult latencyResult;
        private ThroughputTestResult throughputResult;
        private ConcurrencyTestResult concurrencyResult;
        private LoadTestResult loadResult;
        private StressTestResult stressResult;
    }

    @Data
    @Builder
    public static class LatencyTestResult {
        private int testCount;
        private LatencyStats storeLatencies;
        private LatencyStats retrieveLatencies;
        private LatencyStats removeLatencies;
    }

    @Data
    @Builder
    public static class LatencyStats {
        private double average;
        private long min;
        private long max;
        private long p50;
        private long p95;
        private long p99;
    }

    @Data
    @Builder
    public static class ThroughputTestResult {
        private List<ThroughputMeasurement> measurements;
    }

    @Data
    @Builder
    public static class ThroughputMeasurement {
        private String operationType;
        private int operationCount;
        private long durationMs;
        private double throughput;
    }

    @Data
    @Builder
    public static class ConcurrencyTestResult {
        private List<ConcurrencyMeasurement> measurements;
    }

    @Data
    @Builder
    public static class ConcurrencyMeasurement {
        private int threadCount;
        private int operationsPerThread;
        private int totalOperations;
        private long successCount;
        private long errorCount;
        private long durationMs;
        private double throughput;
    }

    @Data
    @Builder
    public static class LoadTestResult {
        private List<LoadMeasurement> measurements;
    }

    @Data
    @Builder
    public static class LoadMeasurement {
        private int connectionCount;
        private long createTimeMs;
        private long queryTimeMs;
        private long randomAccessTimeMs;
        private long actualOnlineCount;
    }

    @Data
    @Builder
    public static class StressTestResult {
        private List<StressMeasurement> measurements;
    }

    @Data
    @Builder
    public static class StressMeasurement {
        private int targetLoad;
        private long actualLoad;
        private long durationMs;
        private boolean success;
        private String errorMessage;
    }
}